/*
 * Copyright 2011 LMAX Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.lmax.disruptor;

import com.lmax.disruptor.util.Util;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.locks.LockSupport;

abstract class SingleProducerSequencerPad extends AbstractSequencer
{
    protected byte
        p10, p11, p12, p13, p14, p15, p16, p17,
        p20, p21, p22, p23, p24, p25, p26, p27,
        p30, p31, p32, p33, p34, p35, p36, p37,
        p40, p41, p42, p43, p44, p45, p46, p47,
        p50, p51, p52, p53, p54, p55, p56, p57,
        p60, p61, p62, p63, p64, p65, p66, p67,
        p70, p71, p72, p73, p74, p75, p76, p77;

    SingleProducerSequencerPad(final int bufferSize, final WaitStrategy waitStrategy)
    {
        super(bufferSize, waitStrategy);
    }
}

abstract class SingleProducerSequencerFields extends SingleProducerSequencerPad
{
    SingleProducerSequencerFields(final int bufferSize, final WaitStrategy waitStrategy)
    {
        super(bufferSize, waitStrategy);
    }

    /**
     * Set to -1 as sequence starting point
     */
    long nextValue = Sequence.INITIAL_VALUE;
    long cachedValue = Sequence.INITIAL_VALUE;
}

/**
 * Coordinator for claiming sequences for access to a data structure while tracking dependent {@link Sequence}s.
 * Not safe for use from multiple threads as it does not implement any barriers.
 *
 * <p>* Note on {@link Sequencer#getCursor()}:  With this sequencer the cursor value is updated after the call
 * to {@link Sequencer#publish(long)} is made.
 */

public final class SingleProducerSequencer extends SingleProducerSequencerFields
{
    protected byte
        p10, p11, p12, p13, p14, p15, p16, p17,
        p20, p21, p22, p23, p24, p25, p26, p27,
        p30, p31, p32, p33, p34, p35, p36, p37,
        p40, p41, p42, p43, p44, p45, p46, p47,
        p50, p51, p52, p53, p54, p55, p56, p57,
        p60, p61, p62, p63, p64, p65, p66, p67,
        p70, p71, p72, p73, p74, p75, p76, p77;

    /**
     * Construct a Sequencer with the selected wait strategy and buffer size.
     *
     * @param bufferSize   the size of the buffer that this will sequence over.
     * @param waitStrategy for those waiting on sequences.
     */
    public SingleProducerSequencer(final int bufferSize, final WaitStrategy waitStrategy)
    {
        super(bufferSize, waitStrategy);
    }

    /**
     * @see Sequencer#hasAvailableCapacity(int)
     */
    @Override
    public boolean hasAvailableCapacity(final int requiredCapacity)
    {
        return hasAvailableCapacity(requiredCapacity, false);
    }

    private boolean hasAvailableCapacity(final int requiredCapacity, final boolean doStore)
    {
        // 判断 nextValue + requiredCapacity是否会覆盖
        long nextValue = this.nextValue;

        long wrapPoint = (nextValue + requiredCapacity) - bufferSize;
        long cachedGatingSequence = this.cachedValue;

        if (wrapPoint > cachedGatingSequence || cachedGatingSequence > nextValue)
        {
            if (doStore)
            {
                cursor.setVolatile(nextValue);  // StoreLoad fence
            }

            long minSequence = Util.getMinimumSequence(gatingSequences, nextValue);
            this.cachedValue = minSequence;

            if (wrapPoint > minSequence)
            {
                return false;
            }
        }

        return true;
    }

    /**
     * @see Sequencer#next()
     */
    @Override
    public long next()
    {
        return next(1);
    }

    /**
     * @see Sequencer#next(int)
     */
    @Override
    public long next(final int n)
    {
        // n表示此次生产者期望获取多少个序号，通常是1
        assert sameThread() : "Accessed by two threads - use ProducerType.MULTI!";

        if (n < 1 || n > bufferSize)
        {
            throw new IllegalArgumentException("n must be > 0 and < bufferSize");
        }

        // 生产者当前序号值
        long nextValue = this.nextValue;

        // 生产者当前序号值+期望获取的序号数量后达到的序号值
        long nextSequence = nextValue + n;

        // 减掉RingBuffer的总的buffer值，用于判断是否出现‘覆盖’
        // 因为是环形数组的设计，所以生产者最多比消费者快一圈
        // 所以减去bufferSize后比较看是否覆盖
        long wrapPoint = nextSequence - bufferSize;

        // 从后面代码分析可得：cachedValue就是“缓存的消费者中”最小序号值，
        // 他“不是当前消费者”中最小序号值，而是上次程序进入到下面的if判定代码段是，被赋值的当时的‘消费者中最小序号值’
        // 这样做的好处在于：在判定是否出现覆盖的时候，不用每次都调用getMininumSequence计算‘消费者中的最小序号值’，从而节约开销。
        // 只要确保当生产者的节奏大于了缓存的cachedGateingSequence一个bufferSize时，从新获取一下 getMinimumSequence()即可。
        long cachedGatingSequence = this.cachedValue;

        // 第一重判断：(wrapPoint > cachedGatingSequence（>=0）) 发生概率较高：
        // 当生产者已经超过上一次缓存的‘消费者中最小序号值’（cachedGatingSequence）一个‘Ring’大小（bufferSize），
        // 需要重新获取cachedGatingSequence，避免出现‘覆盖’
        // 覆盖的情形： 1.当生产者一直在生产，但是消费者不再消费
        // 2. 生产者速度远大于消费者速度

        // 第二重判断：(cachedGatingSequence > nextValue) 发生概率几乎为0 ：
        // 通常情况下 生产者和消费者均为顺序递增的（生产者的seq 和 消费者的seq均不做取余处理，一直递增），
        // 且生产者的seq“先于”消费者的seq，注意是‘先于’而不是‘大于’。
        // 所以不会存在 cachedGatingSequence（消费者消费到的序号） > nextValue（生产者下一个生产的目标序号） 的情况
        // 但是生产者的seq上限为 Long.MAXVALUE
        // 当nextValue>Long.MAXVALUE时，nextValue+n（n>0）就会变成负数，这时候必然会是：cachedGatingSequence > nextValue
        // 【？？？】这个变化的过程会持续bufferSize个序号，这个区间，由于getMinimumSequence()得到的虽然是名义上的‘消费者中最小序号值’，但是不代表是走在‘最后面’的消费者

        if (wrapPoint > cachedGatingSequence || cachedGatingSequence > nextValue)
        {
            // 整个if块都是确保 “覆盖”现象出现的处理策略

            // 将生产者的cursor值更新到主存，以便对所有的消费者线程可见。
            // cursor表示现在生产那个节点了
            cursor.setVolatile(nextValue);  // StoreLoad fence

            long minSequence;

            // 生产者停下来，等待消费者消费，直到‘覆盖’现象清除。
            // 这里会获取此刻所有消费者消费最慢的消费者的消费位移保证不覆盖。
            while (wrapPoint > (minSequence = Util.getMinimumSequence(gatingSequences, nextValue)))
            {// nextValue 的值和 gatingSequences 的值 逐个比较 获取最小的那个

                // 如果无论是那一重判断生效，其实都是为了避免“覆盖现象”
                // 直到获取的消费最慢的消费者的消费序列号>=warpPoint（‘覆盖’现象清除）

                // 通知消费者进行消费，然后自旋
                LockSupport.parkNanos(1L); // TODO: Use waitStrategy to spin?
            }

            // 缓存这一批消费达标（无覆盖现象）的消费者的最小消费序列号
            this.cachedValue = minSequence;
        }

        this.nextValue = nextSequence;

        return nextSequence;
    }

    /**
     * @see Sequencer#tryNext()
     */
    @Override
    public long tryNext() throws InsufficientCapacityException
    {
        return tryNext(1);
    }

    /**
     * @see Sequencer#tryNext(int)
     */
    @Override
    public long tryNext(final int n) throws InsufficientCapacityException
    {
        if (n < 1)
        {
            throw new IllegalArgumentException("n must be > 0");
        }

        if (!hasAvailableCapacity(n, true))
        {
            throw InsufficientCapacityException.INSTANCE;
        }

        // next()与tryNext()方法都会让消费者光标移动 && 返回下一次生产的目标光标 所以调用方法要“做事”
        long nextSequence = this.nextValue += n;

        return nextSequence;
    }

    /**
     * @see Sequencer#remainingCapacity()
     */
    @Override
    public long remainingCapacity()
    {
        long nextValue = this.nextValue;

        long consumed = Util.getMinimumSequence(gatingSequences, nextValue);
        long produced = nextValue;
        // produced - consumed == 剩余未消费的Event数量
        return getBufferSize() - (produced - consumed);
    }

    /**
     * @see Sequencer#claim(long)
     */
    @Override
    public void claim(final long sequence)
    {
        this.nextValue = sequence;
    }

    /**
     * @see Sequencer#publish(long)
     */
    @Override
    public void publish(final long sequence)
    {
        // 将光标设置为当前生产者生产的位置的位置便于消费者消费
        cursor.set(sequence);
        // 通知消费者进行消费

        // BlockingWaitStrategy
        // BusySpinWaitStrategy--->空实现
        // LiteBlockingWaitStrategy
        // LiteTimeoutBlockingWaitStrategy--->
        // PhasedBackoffWaitStrategy
        // SleepingWaitStrategy--->空实现
        // TimeoutBlockingWaitStrategy--->
        // YieldingWaitStrategy--->空实现
        waitStrategy.signalAllWhenBlocking();
    }

    /**
     * @see Sequencer#publish(long, long)
     */
    @Override
    public void publish(final long lo, final long hi)
    {
        // 表示为批量发布事件
        // 但是只关心发布到了那里即可
        // 【？？？】怎么保证不重复消费？
        publish(hi);
    }

    /**
     * @see Sequencer#isAvailable(long)
     */
    @Override
    public boolean isAvailable(final long sequence)
    {
        final long currentSequence = cursor.get();
        return sequence <= currentSequence && sequence > currentSequence - bufferSize;
    }

    @Override
    public long getHighestPublishedSequence(final long lowerBound, final long availableSequence)
    {
        return availableSequence;
    }

    @Override
    public String toString()
    {
        return "SingleProducerSequencer{" +
                "bufferSize=" + bufferSize +
                ", waitStrategy=" + waitStrategy +
                ", cursor=" + cursor +
                ", gatingSequences=" + Arrays.toString(gatingSequences) +
                '}';
    }

    private boolean sameThread()
    {
        return ProducerThreadAssertion.isSameThreadProducingTo(this);
    }

    /**
     * Only used when assertions are enabled.
     */
    private static class ProducerThreadAssertion
    {
        /**
         * Tracks the threads publishing to {@code SingleProducerSequencer}s to identify if more than one
         * thread accesses any {@code SingleProducerSequencer}.
         * I.e. it helps developers detect early if they use the wrong
         * {@link com.lmax.disruptor.dsl.ProducerType}.
         */
        private static final Map<SingleProducerSequencer, Thread> PRODUCERS = new HashMap<>();

        public static boolean isSameThreadProducingTo(final SingleProducerSequencer singleProducerSequencer)
        {
            synchronized (PRODUCERS)
            {
                final Thread currentThread = Thread.currentThread();
                if (!PRODUCERS.containsKey(singleProducerSequencer))
                {
                    PRODUCERS.put(singleProducerSequencer, currentThread);
                }
                return PRODUCERS.get(singleProducerSequencer).equals(currentThread);
            }
        }
    }
}
